package com.mq.rabbitmqtrac.sender;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.rholder.retry.Retryer;
import com.github.rholder.retry.RetryerBuilder;
import com.github.rholder.retry.StopStrategies;
import com.github.rholder.retry.WaitStrategies;
import com.google.common.base.Predicates;
import com.mq.rabbitmqtrac.util.MQConstants;
import com.mq.rabbitmqtrac.util.RabbitMetaMessage;
import com.mq.rabbitmqtrac.util.SpringApplicationContextUtils;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

/**
 * <p><b>Description:</b> RabbitMQ消息发送者
 * <p><b>Company:</b>
 *
 * @version V0.1
 */

@Component
@Slf4j
public class RabbitSender {

    @Autowired
    private RedisTemplate redisTemplate;

    private static Boolean SUCESS_SEND = false;

    /**
     * 发送MQ消息
     *
     * @param rabbitMetaMessage Rabbit元信息对象，用于存储交换器、队列名、消息体
     * @return 消息ID
     * @throws JsonProcessingException
     */
    public String send(RabbitMetaMessage rabbitMetaMessage) throws Exception {
        final String msgId = UUID.randomUUID().toString();

        // 放缓存
        redisTemplate.opsForHash().put(MQConstants.MQ_PRODUCER_RETRY_KEY, msgId, rabbitMetaMessage);
        return sendMsg(rabbitMetaMessage, msgId);
    }

    private String sendMsg(RabbitMetaMessage rabbitMetaMessage, String msgId) throws Exception {
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setMessageId(msgId);
                // 设置消息持久化
                message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                return message;
            }
        };

        ObjectMapper mapper = new ObjectMapper();
        String json = mapper.writeValueAsString(rabbitMetaMessage.getPayload());
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType("application/json");
        Message message = new Message(json.getBytes(), messageProperties);
        try {
            RabbitTemplate rabbitTemplate = SpringApplicationContextUtils.getBean(RabbitTemplate.class);
            rabbitTemplate.convertAndSend(rabbitMetaMessage.getExchange(), rabbitMetaMessage.getRoutingKey(),
                    message, messagePostProcessor, new CorrelationData(msgId));

            log.info("发送消息，消息ID:{}", msgId);

            return msgId;
        } catch (AmqpException e) {
            throw new RuntimeException("发送RabbitMQ消息失败！", e);
        }
    }

    //重发消息
    public void reSendMsg(String msgId, RabbitMetaMessage rabbitMetaMessage) {

        class ReSendThread implements Callable{

            private String msgId;
            private RabbitMetaMessage rabbitMetaMessage;

            private ReSendThread(String msgId, RabbitMetaMessage rabbitMetaMessage) {
                this.msgId = msgId;
                this.rabbitMetaMessage = rabbitMetaMessage;
            }

            @Override
            public Boolean call() throws Exception {
                //如果发送成功，重发结束
                if (SUCESS_SEND)
                    return true;
                //重发消息
                sendMsg(this.rabbitMetaMessage, this.msgId);
                return false;
            }
        }

        Retryer<Boolean> retryer = RetryerBuilder
                .<Boolean>newBuilder()
                //抛出runtime异常、checked异常时都会重试，但是抛出error不会重试。
                .retryIfException()
                .retryIfResult(Predicates.equalTo(false))
                //重试策略  100ms*2^n 最多10s
                .withWaitStrategy(WaitStrategies.exponentialWait(MQConstants.MUTIPLIER_TIME,
                        MQConstants.MAX_RETRY_TIME, TimeUnit.SECONDS))
                .withStopStrategy(StopStrategies.neverStop())
                .build();

        ReSendThread reSendThread = new ReSendThread(msgId, rabbitMetaMessage);
        log.info("重发消息，msgId:{}", msgId);
        try {
            retryer.call(reSendThread);
            //未发送成功，入死信队列
            if(!SUCESS_SEND)
                sendMsgToDeadQueue((String)rabbitMetaMessage.getPayload());
        } catch (Exception e) {
            log.error("重发消息异常");
        }
    }


    //发送到死信队列
    private String sendMsgToDeadQueue(String msg) throws Exception {
        /** 生成一个发送对象 */
        RabbitMetaMessage rabbitMetaMessage = new RabbitMetaMessage();
        /**设置交换机 */
        rabbitMetaMessage.setExchange(MQConstants.DLX_EXCHANGE);
        /**指定routing key */
        rabbitMetaMessage.setRoutingKey(MQConstants.DLX_ROUTING_KEY);
        /** 设置需要传递的消息体,可以是任意对象 */
        rabbitMetaMessage.setPayload(msg);
        /** 发送消息 */
        return send(rabbitMetaMessage);

    }

}
